package cn.doitedu.api;

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/28
 * @Desc: 学大数据，上多易教育
 *
 *   Keyed窗口演示
 **/
public class _25_KeyedWindow_Tumbling_Demo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8899);

        // c1,p01,1000
        SingleOutputStreamOperator<Order> mapped = stream.map(new MapFunction<String, Order>() {
            @Override
            public Order map(String value) throws Exception {

                String[] split = value.split(",");
                return new Order(split[0], split[1], Double.parseDouble(split[2]));
            }
        });


        // 每类商品，每5分钟，的成交金额
        mapped.keyBy(Order::getCategory)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
                .process(new ProcessWindowFunction<Order, JSONObject, String, TimeWindow>() {

                    ValueState<Double> acc;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        acc = getRuntimeContext().getState(new ValueStateDescriptor<Double>("acc", Double.class));
                    }

                    /**
                     * 每次窗口触发，都会根据 key的数量，调用相应次数
                     * 且一次调用，传入的元素都属于同一个key
                     */
                    @Override
                    public void process(String currentKey, ProcessWindowFunction<Order, JSONObject, String, TimeWindow>.Context context, Iterable<Order> elements, Collector<JSONObject> out) throws Exception {

                        String category_id = null;

                        for (Order element : elements) {
                            System.out.println(element);

                            acc.update( acc.value() == null ? element.getAmount() : acc.value() + element.getAmount());

                            category_id = element.getCategory();
                        }

                        System.out.println("---------------------------------");

                        JSONObject resultObject = new JSONObject();
                        resultObject.put("category_id",category_id);
                        resultObject.put("amount",acc.value());
                        resultObject.put("window_start",context.window().getStart());
                        resultObject.put("window_end",context.window().getEnd());

                        out.collect(resultObject);

                    }
                }).print();


        env.execute();
    }

}


@Data
@NoArgsConstructor
@AllArgsConstructor
class Order {
    private String category;
    private String productId;
    private double amount;
}
